/*
 * @Author: wuqingchun
 * @Date: 2024-07-01 13:48:21
 * @Description: 
 * @LastEditors: wuqingchun
 * @LastEditTime: 2024-07-08 20:32:23
 */

#ifndef __ARQ_SENDER_H__
#define __ARQ_SENDER_H__

#include <mutex>
#include <condition_variable>
#include <thread>

#include "arq.h"
#include "arq_socket.h"
#include "arq_allocator.h"
#include "arq_timer.h"

// 窗口控制类模板
template <size_t BUFFER_SIZE = 1024>
class CARQSendWindow
{
    static constexpr size_t BUFFER_INDEX_MASK = BUFFER_SIZE - 1;
    static_assert((BUFFER_SIZE & BUFFER_INDEX_MASK) == 0, "BUFFER_SIZE is not power of 2");
public:
    CARQSendWindow()
    {

    }

    ~CARQSendWindow()
    {
        if (m_ppstUnAckMsg != nullptr)
        {
            delete [] m_ppstUnAckMsg;
            m_ppstUnAckMsg = nullptr;
        }
    }

    int32_t Init(uint32_t uiWndSize)
    {
        if (uiWndSize > BUFFER_SIZE)
        {
            // 设置窗口太大
            return -1;
        }

        m_ppstUnAckMsg = new (std::nothrow) ARQ_HEAD_S*[BUFFER_SIZE];
        if (m_ppstUnAckMsg == nullptr)
        {
            return -1;
        }

        m_uiWndSize = uiWndSize;

        return 0;
    }
    
    bool OnAck(uint32_t uiAckNo)
    {
        if (uiAckNo >= m_uiNextAckNo)
        {
            // 大的消息的AckNo都已经收到，可以确定小的消息也已经收到了
            m_uiNextAckNo = uiAckNo;

            return true;
        }

        return false;
    }

    void AddMsg(ARQ_HEAD_S* pstMsg)
    {
        pstMsg->uiSeqNo = m_uiNextSeqNo++;
        m_ppstUnAckMsg[pstMsg->uiSeqNo & BUFFER_INDEX_MASK] = pstMsg;
    }

    ARQ_HEAD_S* GetMsg(uint32_t uiSeqNo)
    {
        if (uiSeqNo < m_uiNextAckNo || uiSeqNo >= m_uiNextSeqNo)
        {
            return nullptr;
        }

        return m_ppstUnAckMsg[uiSeqNo & BUFFER_INDEX_MASK];
    }

    bool IsWndFull()
    {
        return (m_uiNextSeqNo - m_uiNextAckNo) > m_uiWndSize;
    }

    void GetStatistics()
    {
        printf("Role     : Sender\n");
        printf("WndSize  : %u\n", m_uiWndSize);
        printf("NextSeqNo: %u\n", m_uiNextSeqNo);
        printf("NextAckNo: %u\n", m_uiNextAckNo);
        printf("\n");
    }
    
private:
    ARQ_HEAD_S** m_ppstUnAckMsg = nullptr;
    uint32_t m_uiWndSize = 0;
    uint32_t m_uiNextAckNo = 0;
    uint32_t m_uiNextSeqNo = 0;
};

class CARQSender : public CARQSocket
{
public:
    CARQSender();
    ~CARQSender();

    int32_t Init(SOCKADDR_UN& refPeerAddr, uint32_t uiWndSize);
    int32_t Start();
    int32_t Send(const void* pData, size_t ulLen);

    void* Alloc(size_t ulSize);

    void GetStatistics();

private:
    void __ThreadFuncRecv();

    int32_t __HandleAck(ARQ_HEAD_S& refHdr);

    // 发送心跳消息
    int32_t __SendHeartBeat();

    // 定时任务线程
    void __ThreadFuncTimer();

    // 重传消息
    int32_t __RetransmitMsg(uint32_t uiSeqNo);

private:
    uint32_t m_uiSndWndSize = 16; // 发送窗口
    uint32_t m_uiNextSeqNo = 0;
    uint32_t m_uiNextAckNo = 0;

    std::thread m_objRecvThread; // 接收线程
    volatile bool m_bRecvRuning = false;

    CARQAllocator m_objAllocator;

    std::mutex m_objMtxSnd;
    std::condition_variable m_objCVSnd;

    CARQTimer m_objTimer;
    std::thread m_objTimerThread; // 定时线程
    volatile bool m_bTimerRuning = false;

    CARQSendWindow<> m_objSndWnd;
};

#endif //__ARQ_SENDER_H__